#include "zrpc/net/tcp_client.h"

#include "zrpc/net/tcp_connection.h"

namespace zrpc {
TcpClient::TcpClient(EventLoop *loop, const std::string &ip, int16_t port,
                     const std::any &context)
    : connector_(new Connector(loop, ip.c_str(), port, true)),
      loop_(loop),
      context_(context),
      ip_(ip),
      port_(port),
      retry_(false),
      connecting_(true),
      connection_(nullptr) {
  connector_->SetNewConnectionCallback(
      std::bind(&TcpClient::NewConnection, this, std::placeholders::_1));
  connector_->SetConnectionErrorCallBack(
      std::bind(&TcpClient::ErrorConnection, this));
}

namespace detail {
void RemoveConnection(EventLoop *loop,
                      const std::shared_ptr<TcpConnection> &conn) {
  loop->QueueInLoop(std::bind(&TcpConnection::ConnectDestroyed, conn));
}

void RemoveConnector(const std::shared_ptr<Connector> &connector_) {}
}  // namespace detail

TcpClient::~TcpClient() {
  std::shared_ptr<TcpConnection> conn;
  bool unique = false;
  {
    std::unique_lock<std::mutex> lk(mutex_);
    unique = connection_.unique();
    conn = connection_;
  }

  if (conn) {
    assert(loop_ == conn->GetLoop());
    CloseCallback cb = std::bind(&detail::RemoveConnection, loop_, conn);
    loop_->RunInLoop(std::bind(&TcpConnection::SetCloseCallback, conn, cb));
    if (unique) {
      conn->ForceClose();
    }
  } else {
    connector_->Stop();
    loop_->RunAfter(1, false, std::bind(&detail::RemoveConnector, connector_));
  }
}

void TcpClient::EnableRetry() {
  connector_->EnableRetry();
  retry_ = true;
}

void TcpClient::CloseRetry() {
  connector_->CloseRetry();
  retry_ = false;
}

void TcpClient::Connect(bool s) {
  connecting_ = true;
  connector_->Start(s);
}

void TcpClient::DisConnect() {
  connecting_ = false;
  {
    std::unique_lock<std::mutex> lk(mutex_);
    if (connection_) {
      connection_->Shutdown();
    }
  }
}

std::shared_ptr<TcpConnection> TcpClient::GetConnection() {
  std::unique_lock<std::mutex> lk(mutex_);
  return connection_;
}

void TcpClient::Stop() {
  connecting_ = false;
  connector_->Stop();
}

void TcpClient::ErrorConnection() { connection_error_callBack_(); }

void TcpClient::NewConnection(int32_t sockfd) {
  loop_->AssertInLoopThread();
  std::shared_ptr<TcpConnection> conn(
      new TcpConnection(loop_, sockfd, context_));
  conn->SetConnectionCallback(std::move(connection_callback_));
  conn->SetMessageCallback(std::move(message_callback_));
  conn->SetWriteCompleteCallback(std::move(write_complete_callback_));
  conn->SetCloseCallback(
      std::bind(&TcpClient::RemoveConnection, this, std::placeholders::_1));

  {
    std::unique_lock<std::mutex> lk(mutex_);
    connection_ = conn;
  }
  conn->ConnectEstablished();
}

void TcpClient::RemoveConnection(const std::shared_ptr<TcpConnection> &conn) {
  loop_->AssertInLoopThread();
  assert(loop_ == conn->GetLoop());
  {
    std::unique_lock<std::mutex> lk(mutex_);
    assert(connection_ == conn);
    connection_.reset();
  }

  loop_->QueueInLoop(std::bind(&TcpConnection::ConnectDestroyed, conn));
  if (retry_ && connecting_) {
    connector_->Restart();
  }
}
}  // namespace zrpc